package com.burges.net.metrics

import org.apache.flink.api.common.functions.RichMapFunction
import org.apache.flink.api.scala.metrics.ScalaGauge
import org.apache.flink.configuration.Configuration
import org.apache.flink.dropwizard.metrics.{DropwizardHistogramWrapper, DropwizardMeterWrapper}
import org.apache.flink.metrics.{Counter, Histogram, Meter}

/**
  * 创建人    BurgessLee 
  * 创建时间   2020/2/28 
  * 描述     四种类型用户自定义的监控指标示例代码
  */
object MetricsDemo {

	def main(args: Array[String]): Unit = {

	}

	//Counters指标
	class MyMapper extends RichMapFunction[Long, Long]{
		@transient private var counter: Counter = _

		// Open方法中获取Counter实例化对象
		override def open(parameters: Configuration): Unit = {
			counter = getRuntimeContext.getMetricGroup.counter("MyCounter")
		}

		override def map(value: Long): Long = {
			if(value > 0){
				counter.inc()
			}
			value
		}
	}

	//Gauges指标
	class gaugeMapper extends RichMapFunction[String, String]{
		@transient private var countValue = 0

		//在Open方法中获取gauge实例化对象
		override def open(parameters: Configuration): Unit = {
			getRuntimeContext.getMetricGroup.gauge[Int, ScalaGauge[String]]("MyGague", ScalaGauge[Int] (() => countValue))
		}

		override def map(value: String): String = {
			countValue += 1  //累加
			value
		}
	}

	//Higstograms指标
	class HistogramMapper extends RichMapFunction[Long, Long] {
		@transient private var histogram: Histogram = _

		//在Open方法中获取Histogram对象
		override def open(parameters: Configuration): Unit = {
			//使用DropwizardHistogramWrapper包装类转换Codahale/DropWizard Histograms
			val dropwizardHistogram = new com.codahale.metrics.Histogram(new SildingWindowReservoir(500))
			histogram = getRuntimeContext.getMetricGroup.histogram("myHistogram", new DropwizardHistogramWrapper(dropwizardHistogram))
		}

		override def map(value: Long): Long = {
			histogram.update(value)
			value
		}
	}

	//Meters指标
	class MetersMapper extends RichMapFunction[Long, Long]{

		@transient private var meter: Meter = _
		//使用DropwizardMeterWrapper包装类转换Codahale/DropWizard Meter
		override def open(parameters: Configuration): Unit = {
			val dropwizardMeter = new com.codahale.metrics.Meter()
			meter = getRuntimeContext.getMetricGroup.meter("myMeter", new DropwizardMeterWrapper(dropwizardMeter))
		}

		override def map(in: Long): Long = {
			//更新指标
			meter.markEvent()
			in
		}
	}

}
